package qdu.edu.cn.storm.bolt;

import org.apache.storm.topology.BasicOutputCollector;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseBasicBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;

import static qdu.edu.cn.storm.ConstantFinalString.Fields.*;

public class SplitWordsBolt extends BaseBasicBolt {

    @Override
    public void execute(Tuple input, BasicOutputCollector collector) {
        // 由于使用DRPC，无法在Spout中设置字段名称，所以需要使用序号
        // D-RPC：storm中的bolt发射的第一个字段必须是request-id，每个bolt都要取上一个bolt组件发射来的第一个字段，然后原样发射出去，最后一个bolt只能发射两个字段，一个是request-id，另一个是结果

        String[] words = splitWords(input.getString(1));
        for (String word : words) {
            collector.emit(new Values(input.getValue(0), word));
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {
        declarer.declare(new Fields(USER_ID, WORD));
    }

    private String[] splitWords(String sentence) {
        // ("abcdef", 2) ==> (ab, bc, cd, de, ef)
        Set<String> wordsSet = new HashSet<>(Arrays.asList(sentence.split("")));

        for (int length = sentence.length()-1; length > 1; length--) {
            for (int i = 0; i <= sentence.length() - length; i++) {
                wordsSet.add(sentence.substring(i, i + length)); // beginLength, endLength
            }
        }
        wordsSet.add(sentence);

        return wordsSet.toArray(new String[0]);
    }
}
